
#include <readline/readline.h>

#include <iostream>

#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/tokenizer.hpp>

#include <xbase_utility.hpp>
#include <tablet_name_parser.hpp>

#include "table_info_cache.hpp"
#include "row_mutator.hpp"
#include "session_manager.hpp"
#include "protocol.hpp"
#include "sync_handler.hpp"
#include "io_service_pool_impl.hpp"

using namespace xbase;

boost::mutex m;
std::size_t pending_ops;
std::deque<std::string> complete_info;

boost::asio::ip::tcp::endpoint master_ep;

io_service_pool_ptr s_pool;
session_manager *manager = NULL;
location_cache *cache = NULL;
table_info_cache * the_table_info_cache = NULL;

void on_apply(row_mutation mutation,
	      uint64_t start_ts,
	      const boost::system::error_code &ec)
{
	uint64_t end_ts = cur_micro_seconds();
	double used = (end_ts - start_ts);
	used /= (1000.0 * 1000.0);
	std::ostringstream out;
	out << "---" << std::endl;
	out << "[" << dump_timestamp(start_ts)
	    << " - " << dump_timestamp(end_ts)
	    << "] used " << used << " seconds" << std::endl;
	out << "written row \""
	    << mutation.row_name()
	    << "\" into table \"" << mutation.table_name()
	    << "\", error_code: "
	    << ec << " - " << ec.message() << std::endl;
	boost::mutex::scoped_lock lock(m);
	complete_info.push_back(out.str());
}

void show_data(const row_mutation_data &data)
{
	std::cout << data.to_string() << std::endl;
}

void command_error() 
{
	std::cout << "error: bad command." << std::endl
		  << "command list:" << std::endl
		  << "\tadd <table name> <row> <column> [value] [timestamp]" << std::endl
		  << "\tapply [table name] [row] [timestamp]" << std::endl
		  << "\tquit" << std::endl;
}

template<typename F>
void start_xbase(F f);

void user_main(int argc, char **argv);

int main(int argc, char **argv)
{
	if(argc != 3)
	{
		std::cout << "test_row_mutator <master ip> <master port>" << std::endl;
		return 0;
	}
	
	boost::asio::ip::tcp::endpoint mep(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);
	
	master_ep = mep;

	start_xbase(boost::bind(user_main, argc, argv));
	return 0;
}

template<typename F>
void start_xbase(F f)
{
	// init
	
	LOG_CONFIG("../../conf/log4cplus.tablet_server.conf");

	async_module_stats::enable();

	io_service_pool_ptr pool(new impl::single_io_service());
	client_options options;
	pool->set_options(options);
	s_pool = pool;
	pool->run();
	session_manager sm(*pool);
	table_info_cache tic(*pool, master_ep, sm);
	root_meta_cache rmc(*pool, master_ep, sm);
	location_cache lc(*pool, rmc, sm);
	cache = &lc;
	manager = &sm;
	the_table_info_cache = &tic;

	// -------

	f();

	// -------

	// close
	
	sm.close();
	pool->stop();
	pool->join();

	async_module_stats::print(std::cout);
	std::cout << std::endl;
}


void user_main(int argc, char **argv)
{
	typedef boost::tokenizer<boost::escaped_list_separator<char> > tok;
	boost::escaped_list_separator<char> separator("\\", ", ", "\"");

	typedef std::map<std::string, row_mutation> row_map;
	typedef boost::shared_ptr<row_map> row_map_ptr;
	typedef std::pair<protocol::table_info_ptr, row_map_ptr> table_item;
	typedef std::map<std::string, table_item> table_map;
	table_map tables;
	while(true)
	{
		{
			boost::mutex::scoped_lock lock(m);
			while(! complete_info.empty())
			{
				std::cout << complete_info.front();
				complete_info.pop_front();
				--pending_ops;
			}
		}

		char *input = readline("> ");
			
		if(input == NULL)
		{
			break;
		}
		const std::string copy = input;
		free(input);
		if((copy.empty()) ||
		   (copy[0] == '#')) // 注释
		{
			continue;
		}
		
		tok t(copy, separator);

		// for(tok::iterator it = t.begin();
		//     it != t.end(); ++it)
		// 	std::cout << "\"" << *it << "\",";
		// std::cout << std::endl;

		tok::iterator it = t.begin();
		if(it == t.end())
		{
			;
		}
		else if(*it == "quit")
		{
			break;
		}
		else if(*it == "add")
		{
			std::string table_name;
			std::string row;
			std::string column;
			std::string value;
			std::string timestamp;

			++it;
			if(it != t.end())
			{
				table_name = *it;
				++it;
			}
			if(it != t.end())
			{
				row = *it;
				++it;
			}
			if(it != t.end())
			{
				column = *it;
			}
			// 上面是必写的
			if(it == t.end())
			{
				command_error();
				continue;
			}
			// 这是可选的
			++it;
			if(it != t.end())
			{
				value = *it;
				++it;
			}
			if(it != t.end())
			{
				timestamp = *it;
			}

			if(table_name.empty())
			{
				std::cout << "error: " << error::bad_table_name << std::endl;
				continue;
			}
			if(row.empty())
			{
				std::cout << "error: " << error::bad_row_name << std::endl;
				continue;
			}

			table_map::iterator it_table = tables.find(table_name);
			if(it_table == tables.end())
			{
				// 查该表的表结构信息
				protocol::table_info_ptr table_info;
				table_info_cache::sync_handler_type sh;
				std::cout << "query table info for " << table_name << " ... " << std::endl;
				the_table_info_cache->async_get_table_info(table_name,
									   sh.make_handler());
				sh.wait();
				if(sh.get_error_code())
				{
					std::cout << "error: " << sh.get_error_code() << " - " << sh.get_error_code().message() << std::endl;
					return;
				}
				table_info = sh.get_table_info();
				std::cout << "got table info: " << table_info->to_string() << std::endl;

				table_item item(table_info, row_map_ptr(new row_map()));
				
				std::pair<table_map::iterator, bool> ret =
					tables.insert(table_map::value_type(table_name, item));
				assert(ret.second == true);
				it_table = ret.first;
			}
			row_map &rows = *(it_table->second.second);
			row_map::iterator it = rows.find(row);
			if(it == rows.end())
			{
				std::pair<row_map::iterator, bool> ret =
					rows.insert(row_map::value_type(row, row_mutation(it_table->second.first, row)));
				assert(ret.second == true);
				it = ret.first;
			}
			row_mutation &mutation = it->second;
			
			uint64_t ts;
			try {
				ts = boost::lexical_cast<uint64_t>(timestamp);
			} catch(const std::exception &e) {
				ts = 0;
			}
			boost::system::error_code ec =
				mutation.add(column, value, ts);
			if(ec)
			{
				std::cout << "add failed: " << ec << " - " << ec.message() << std::endl;
			}
		}
		else if(*it == "apply")
		{
			std::string table_name;
			std::string row;
			std::string timestamp;

			// ++it;
			// if(it != t.end())
			// {
			// 	table_name = *it;
			// }
			// // 上面是必写的
			// if(it == t.end())
			// {
			// 	command_error();
			// 	continue;
			// }
			// 下面是可选的
			++it;
			if(it != t.end())
			{
				table_name = *it;
				++it;
			}
			if(it != t.end())
			{
				row = *it;
				++it;
			}
			if(it != t.end())
			{
				timestamp = *it;
			}

			std::deque<table_map::value_type> row_queue;

			if(table_name.empty()) // apply all table
			{
				for(table_map::iterator it = tables.begin();
				    it != tables.end(); ++it)
				{
					row_queue.push_back(*it);
				}
				tables.clear();
			}
			else
			{
				const table_map::iterator it = tables.find(table_name);
				if(it != tables.end())
				{
					row_queue.push_back(*it);
					tables.erase(it);
				}
			}

			std::deque<row_mutation> ptr_q;

			while(! row_queue.empty())
			{
				table_map::value_type item = row_queue.front();
				row_queue.pop_front();

				row_map &rows = *item.second.second;

				if(row.empty()) // apply all row
				{
					for(row_map::iterator it = rows.begin();
					    it != rows.end(); ++it)
					{
						ptr_q.push_back(it->second);
					}
					rows.clear();
				}
				else
				{
					const row_map::iterator it = rows.find(row);
					if(it != rows.end())
					{
						ptr_q.push_back(it->second);
						rows.erase(it);
					}
				}

				if(! item.second.second->empty()) // 如果数据还有剩余，则再放回到table_map中
				{
					std::pair<table_map::iterator, bool> ret =
						tables.insert(item);
					assert(ret.second == true);
				}
			}

			if(ptr_q.empty())
			{
				std::cout << "not data"
					  << (row.empty() ? "" : " for row \"")
					  << row << '"'
					  << (table_name.empty() ? "" : " in table \"")
					  << table_name << '"'
					  << std::endl;
			}

			while(! ptr_q.empty())
			{
				row_mutation mutation = ptr_q.front();
				ptr_q.pop_front();

				show_data(mutation.data());

				if(mutation.empty())
				{
					std::cout << "no mutations for row \""
						  << mutation.row_name()
						  << "\" in table \""
						  << mutation.table_name() << '"'
						  << std::endl;
					continue;
				}

				uint64_t ts;
				try {
					ts = boost::lexical_cast<uint64_t>(timestamp);
				} catch(const std::exception &e) {
					ts = cur_micro_seconds();
				}

				boost::mutex::scoped_lock lock(m);
				++pending_ops;
				lock.unlock();

				uint64_t start_ts = cur_micro_seconds();
				async_apply(*s_pool, *manager, *cache,
					    mutation,
					    ts,
					    boost::bind(on_apply,
							mutation, start_ts,
							_1));
			}
		}
		else
		{
			command_error();
		}
	}

	boost::mutex::scoped_lock lock(m);
	while(pending_ops > 0)
	{
		while(! complete_info.empty())
		{
			std::cout << complete_info.front();
			complete_info.pop_front();
			--pending_ops;
		}
		if(pending_ops == 0)
			break;
		lock.unlock();
		async_module_stats::print(std::cout);
		std::cout << std::endl;
		std::cout << "wait pending ops "<< pending_ops <<" ..." << std::endl;
		sleep(1);
		lock.lock();
	}

}
